#define LOG_TAG "web_socket_server"
#include <stdio.h>

#include <condition_variable>
#include <iostream>
#include <mutex>
#include <string>
#include <thread>

#define _USE_MATH_DEFINES
#include <math.h>

#include "esp_event.h"
#include "esp_system.h"
#include "esp_wifi.h"
#include "nvs_flash.h"
// #include "protocol_examples_common.h"

#include "Trace.h"
#include "WebSocketService.h"
#include "c3log.h"
#include "esp_event.h"
#include "esp_log.h"
#include "esp_websocket_client.h"
#include "freertos/FreeRTOS.h"
#include "freertos/event_groups.h"
#include "freertos/semphr.h"
#include "freertos/task.h"
#include "power_track/Ina226App.h"
#include "websocket_server.h"
#include <WifiManager.h>
#include "../WebService.h"

#undef LOG_TAG
#define LOG_TAG "web_socket_server"

WebSocketService* WebSocketService::m_pInstance = nullptr;
bool gWebSocketConn = false;
static QueueHandle_t client_queue;
static QueueHandle_t clients_request_msg_queue;
const static int client_queue_size = 10;
int root_client = 0;
#define MSG_SIZE (1024)
struct ws_message {
    uint8_t msg_out[MSG_SIZE];
    int msg_out_size = 0;
};
int pp_id = 0;  // pingpangid
ws_message ws_msg[2];

uint8_t poll_data[MSG_SIZE];
uint32_t poll_data_pos = 0;

int clientList[WEBSOCKET_SERVER_MAX_CLIENTS] = {0};  // 1: on 0 :off

// handles websocket events
void websocket_process_request_callback(uint8_t num, WEBSOCKET_TYPE_t type,
                                        char* msg, uint64_t len) {
    const static char* TAG = "websocket_process_request_callback";
    int value = 0;

    switch (type) {
        case WEBSOCKET_CONNECT:
            gWebSocketConn = true;
            ESP_LOGI(TAG, "client %i connected!", num);
            break;
        case WEBSOCKET_DISCONNECT_EXTERNAL:
            gWebSocketConn = false;
            ESP_LOGI(TAG, "client %i sent a disconnect message", num);
            break;
        case WEBSOCKET_DISCONNECT_INTERNAL:
            gWebSocketConn = false;
            ESP_LOGI(TAG, "client %i was disconnected", num);
            break;
        case WEBSOCKET_DISCONNECT_ERROR:
            gWebSocketConn = false;
            ESP_LOGI(TAG, "client %i was disconnected due to an error", num);
            break;
        case WEBSOCKET_TEXT:
            if (len) {  // if the message length was greater than zero
                // ClientsMsg post
                ClientsMsg* pClientMsg = new ClientsMsg();
                pClientMsg->clientId = num;
                memcpy(pClientMsg->pMsg, msg,
                       len > MAX_REQUEST_STRING_SIZE ? MAX_REQUEST_STRING_SIZE
                                                     : len);
                pClientMsg->msgLen = len;
                portBASE_TYPE ret = xQueueSendToBack(
                    clients_request_msg_queue, &pClientMsg, portMAX_DELAY);
                if (ret == pdTRUE) {
                    ESP_LOGI(TAG, "[cb]request_msg recieved and dispatch", num);
                } else {
                    delete pClientMsg;
                    pClientMsg = nullptr;
                }

                // process
                switch (msg[0]) {
                    case 'L':
                        if (sscanf(msg, "L%i", &value)) {
                            ESP_LOGI(TAG, "LED value: %i", value);
                            // todo led_duty(value);
                            if (gWebSocketConn) {
                                ws_server_send_text_all_from_callback(
                                    msg, len);  // broadcast it!
                                ESP_LOGI(TAG, "start L task");
                                memset(ws_msg[pp_id].msg_out, 1, MSG_SIZE - 1);
                                ws_msg[pp_id].msg_out[MSG_SIZE - 1] = 0;
                                // for (int i = 0; i < 10000; i++) {
                                // ws_server_send_text_client_from_callback(root_client,
                                // msg_out, MSG_SIZE); vTaskDelay(1000 /
                                // portTICK_PERIOD_MS);
                                // }
                            }
                            ESP_LOGI(TAG, "end L task");
                        }
                        break;
                    case 'C':
                        if (sscanf(msg, "C%i", &value)) {
                            ESP_LOGI(TAG, "value: %i", value);
                            ESP_LOGI(TAG, "start Current Sample task");
                            memset(&ws_msg[pp_id].msg_out, 1, MSG_SIZE - 1);
                            ws_msg[pp_id].msg_out[MSG_SIZE - 1] = 0;
                            for (int i = 0; i < 10000; i++) {
                                if (gWebSocketConn) {
                                    ws_server_send_text_client_from_callback(
                                        root_client, (char*)ws_msg[pp_id].msg_out,
                                        MSG_SIZE);
                                }
                            }
                            ESP_LOGI(TAG, "end L task");
                        }
                        break;
                    case 'M':
                        ESP_LOGI(TAG, "got message length %i: %s", (int)len - 1,
                                 &(msg[1]));
                        break;
                    default:
                        ESP_LOGI(TAG, "got an unknown message with length %i",
                                 (int)len);
                        break;
                }
            }
            break;
        case WEBSOCKET_BIN:
            ESP_LOGI(TAG, "client %i sent binary message of size %i:\n%s", num,
                     (uint32_t)len, msg);
            break;
        case WEBSOCKET_PING:
            ESP_LOGI(TAG, "client %i pinged us with message of size %i:\n%s",
                     num, (uint32_t)len, msg);
            break;
        case WEBSOCKET_PONG:
            ESP_LOGI(TAG, "client %i responded to the ping", num);
            break;
    }
}

// serves any clients
static void create_http_server(struct netconn* conn, void* pvParameters,
                               void (*callback)(uint8_t num,
                                                WEBSOCKET_TYPE_t type,
                                                char* msg, uint64_t len)) {
    const static char* TAG = "http_server";
    const static char HTML_HEADER[] =
        "HTTP/1.1 200 OK\nContent-type: text/html\n\n";
    const static char ERROR_HEADER[] =
        "HTTP/1.1 404 Not Found\nContent-type: text/html\n\n";
    const static char JS_HEADER[] =
        "HTTP/1.1 200 OK\nContent-type: text/javascript\n\n";
    const static char CSS_HEADER[] =
        "HTTP/1.1 200 OK\nContent-type: text/css\n\n";
    // const static char PNG_HEADER[] = "HTTP/1.1 200 OK\nContent-type:
    // image/png\n\n";
    const static char ICO_HEADER[] =
        "HTTP/1.1 200 OK\nContent-type: image/x-icon\n\n";
    // const static char PDF_HEADER[] = "HTTP/1.1 200 OK\nContent-type:
    // application/pdf\n\n"; const static char EVENT_HEADER[] = "HTTP/1.1 200
    // OK\nContent-Type: text/event-stream\nCache-Control: no-cache\nretry:
    // 3000\n\n";
    struct netbuf* inbuf;
    static char* buf;
    static uint16_t buflen;
    static err_t err;

    netconn_set_recvtimeout(conn,
                            1000);  // allow a connection timeout of 1 second
    ESP_LOGI(TAG, "reading from client...");
    err = netconn_recv(conn, &inbuf);
    ESP_LOGI(TAG, "read from client");
    if (err == ERR_OK) {
        netbuf_data(inbuf, (void**)&buf, &buflen);
        if (buf) {
            // default page
            if (strstr(buf, "GET / ") && !strstr(buf, "Upgrade: websocket")) {
                ESP_LOGI(TAG, "Sending /");
                netconn_write(conn, HTML_HEADER, sizeof(HTML_HEADER) - 1,
                              NETCONN_NOCOPY);
                // netconn_write(conn,
                // root_html_start,root_html_len,NETCONN_NOCOPY);
                netconn_close(conn);
                netconn_delete(conn);
                netbuf_delete(inbuf);
            }

            // default page websocket
            else if (strstr(buf, "GET / ") &&
                     strstr(buf, "Upgrade: websocket")) {
                ESP_LOGI(TAG, "Requesting websocket on /");
                root_client =
                    ws_server_add_client(conn, buf, buflen, "/", callback);
                clientList[root_client] = 1;
                netbuf_delete(inbuf);
            }

            else if (strstr(buf, "GET /test.js ")) {
                ESP_LOGI(TAG, "Sending /test.js");
                netconn_write(conn, JS_HEADER, sizeof(JS_HEADER) - 1,
                              NETCONN_NOCOPY);
                // netconn_write(conn, test_js_start,
                // test_js_len,NETCONN_NOCOPY);
                netconn_close(conn);
                netconn_delete(conn);
                netbuf_delete(inbuf);
            }

            else if (strstr(buf, "GET /test.css ")) {
                ESP_LOGI(TAG, "Sending /test.css");
                netconn_write(conn, CSS_HEADER, sizeof(CSS_HEADER) - 1,
                              NETCONN_NOCOPY);
                // netconn_write(conn, test_css_start,
                // test_css_len,NETCONN_NOCOPY);
                netconn_close(conn);
                netconn_delete(conn);
                netbuf_delete(inbuf);
            }

            else if (strstr(buf, "GET /favicon.ico ")) {
                ESP_LOGI(TAG, "Sending favicon.ico");
                netconn_write(conn, ICO_HEADER, sizeof(ICO_HEADER) - 1,
                              NETCONN_NOCOPY);
                // netconn_write(conn,favicon_ico_start,favicon_ico_len,NETCONN_NOCOPY);
                netconn_close(conn);
                netconn_delete(conn);
                netbuf_delete(inbuf);
            }

            else if (strstr(buf, "GET /")) {
                ESP_LOGI(TAG, "Unknown request, sending error page: %s", buf);
                netconn_write(conn, ERROR_HEADER, sizeof(ERROR_HEADER) - 1,
                              NETCONN_NOCOPY);
                // netconn_write(conn, error_html_start,
                // error_html_len,NETCONN_NOCOPY);
                netconn_close(conn);
                netconn_delete(conn);
                netbuf_delete(inbuf);
            } else {
                ESP_LOGI(TAG, "Unknown request");
                netconn_close(conn);
                netconn_delete(conn);
                netbuf_delete(inbuf);
            }
        } else {
            ESP_LOGI(TAG, "Unknown request (empty?...)");
            netconn_close(conn);
            netconn_delete(conn);
            netbuf_delete(inbuf);
        }
    } else {  // if err==ERR_OK
        ESP_LOGI(TAG, "error on read, closing connection");
        netconn_close(conn);
        netconn_delete(conn);
        netbuf_delete(inbuf);
    }
}

// interface

// handles clients when they first connect. passes to a queue
static void create_ws_server_task(void* pvParameters) {
    // WebSocketService* serv = (WebSocketService*)pvParameters;
    // const static char* TAG = "server_task";
    struct netconn *conn, *newconn;
    static err_t err;
    client_queue = xQueueCreate(client_queue_size, sizeof(struct netconn*));

    conn = netconn_new(NETCONN_TCP);
    netconn_bind(conn, NULL, 80);
    netconn_listen(conn);
    LOGI("server listening");
    do {
        err = netconn_accept(conn, &newconn);
        LOGI("new client");
        if (err == ERR_OK) {
            // post task to "create_session_task"
            xQueueSendToBack(client_queue, &newconn, portMAX_DELAY);
        }
    } while (err == ERR_OK);
    netconn_close(conn);
    netconn_delete(conn);
    LOGI("task ending, rebooting board");
    esp_restart();
    vTaskDelete(NULL);
}

// receives clients from queue, handles them
static void create_ws_session_task(void* pvParameters) {
    WebSocketService* serv = (WebSocketService*)pvParameters;
    struct netconn* conn;
    LOGI("task starting");
    for (;;) {
        xQueueReceive(client_queue, &conn, portMAX_DELAY);
        LOGI("conn session creating");
        if (!conn) continue;
        create_http_server(conn, (void*)serv,
                           websocket_process_request_callback);
    }
    vTaskDelete(NULL);
}

// Steam Socket
SemaphoreHandle_t xPollingStartSemaphore;
bool startPoll = false;
SemaphoreHandle_t xSemaphore;

SemaphoreHandle_t xPollDataSemaphore;
static BaseType_t xHigherPriorityTaskWoken;

void process_request_task(void* pvParameters) {
    WebSocketService* serv = (WebSocketService*)pvParameters;
    const static char* TAG = "process_request_task";
    struct ClientsMsg* pClientMsg = nullptr;
    ESP_LOGI(TAG, "process_request_task starting");

    for (;;) {
        xQueueReceive(clients_request_msg_queue, &pClientMsg, portMAX_DELAY);
        ESP_LOGI(TAG, "process_request_task: rec qequest:%p", pClientMsg);
        if (nullptr == pClientMsg) continue;
        // process
        ESP_LOGI(TAG, "msg: %s", pClientMsg->pMsg);

        // setRepeatRequest
        int cmd = 0;
        if (sscanf((char*)pClientMsg->pMsg, "setRepeatRequest: %i", &cmd)) {
            // Stream Request
            ESP_LOGI(TAG, "Stream Request Cmd: %i", cmd);
            startPoll = true;
            xSemaphoreGiveFromISR(xPollingStartSemaphore,
                                  &xHigherPriorityTaskWoken);
        } else if (sscanf((char*)pClientMsg->pMsg, "stopRepeatRequest: %i",
                          &cmd)) {
            // Stream Request
            ESP_LOGI(TAG, "Stream Request Cmd: %i", cmd);
            startPoll = false;
            xSemaphoreGiveFromISR(xPollingStartSemaphore,
                                  &xHigherPriorityTaskWoken);
        }
        delete pClientMsg;
        pClientMsg = nullptr;
    }
    vTaskDelete(NULL);
}

// polling thread
static void polling_task(void* pvParameters) {
    WebSocketService* serv = (WebSocketService*)pvParameters;
    // const static char* TAG = "polling_task";
    LOGI("polling task starting");
    Ina226App inaApp;
    inaApp.setup();
    // wait and polling Client
    for (;;) {
        BaseType_t rec =
            xSemaphoreTake(xPollingStartSemaphore,
                           pdMS_TO_TICKS(2 * 1000));  // wait client connect
        if (rec != pdTRUE || !startPoll) {
            // LOGI("polling_task waiting %d:%d", rec!=pdTRUE, !startPoll);
            continue;
        }

#if 0
    // polling data
    rec = xSemaphoreTake(xPollDataSemaphore, pdMS_TO_TICKS(1000));//pdMS_TO_TICKS将ms转成对应的滴答数
    if (rec==pdTRUE) {
        LOGI("rec and dispatch msg start");
        memcpy(ws_msg[pp_id].msg_out, poll_data, MSG_SIZE);
        ws_msg[pp_id].msg_out_size = MSG_SIZE;
        pp_id = (pp_id + 1) % 2;  // switch to ping
       xSemaphoreGiveFromISR(xSemaphore, &xHigherPriorityTaskWoken);
    }

#else   // FAST PULL
        // gene data
        // static uint64_t g_degree = 0;
        // float cur, ref, power_ref;

        int curIdxOffset = 0;
        // if (WEB_SOCKET_STARTED == powerApp->mpWebSocketServer->getStatus()) {
        for (; startPoll && gWebSocketConn;) {
            if (curIdxOffset < (MSG_SIZE - sizeof(float) * 3)) {  // 1000 f/s
                float* curPtr = (float*)(ws_msg[pp_id].msg_out + curIdxOffset);
                inaApp.pollIVP();
                inaApp.getIVP(curPtr[0], curPtr[1], curPtr[2]);
                curIdxOffset += sizeof(float) * 3;
                // vTaskDelay(5 / portTICK_PERIOD_MS);
            } else {
                ws_msg[pp_id].msg_out_size = curIdxOffset;
                pp_id = (pp_id + 1) % 2;  // switch to ping
                curIdxOffset = 0;
                BaseType_t lxHigherPriorityTaskWoken =
                    pdTRUE;                  // force switch context
                xSemaphoreGive(xSemaphore);  // 10f/s
                // vTaskDelay(1 / portTICK_PERIOD_MS);
                LOGI("polling_task: data prepared");
            }
        }
#endif
    }
    vTaskDelete(NULL);
}

// stream_cb_task thread
static void stream_cb_task(void* pvParameters) {
    WebSocketService* serv = (WebSocketService*)pvParameters;
    const static char* TAG = "dispatch_task";
    LOGI(TAG, "dispatch_task starting");
    for (;;) {
        BaseType_t msToWait = 1000;  // 信号量接收等待时间
        BaseType_t rec = xSemaphoreTake(
            xSemaphore,
            pdMS_TO_TICKS(msToWait));  // pdMS_TO_TICKS将ms转成对应的滴答数
        if (rec == pdTRUE && gWebSocketConn) {
            LOGI(TAG, "dispatch msg start");
            ws_server_send_bin_client_from_callback(
                root_client, (char*)ws_msg[(pp_id + 1) % 2].msg_out,
                ws_msg[(pp_id + 1) % 2].msg_out_size);  // pang
        }
    }
    vTaskDelete(NULL);
}

// stream_cb_task thread
static void stream_cb_immed_task(void* pvParameters) {
    WebSocketService* serv = (WebSocketService*)pvParameters;
    const static char* TAG = "dispatch_immed_task";
    for (;;) {
        vTaskDelay(1000 / portTICK_PERIOD_MS);
        if (!gWebSocketConn) {
            ws_server_send_bin_client_from_callback(
                root_client, (char*)ws_msg[(pp_id + 1) % 2].msg_out,
                ws_msg[(pp_id + 1) % 2].msg_out_size);  // pang
        }
    }
    vTaskDelete(NULL);
}

// rigiseter
int WebSocketService::registerStreamSocket(void* streamSrc) {
    clients_request_msg_queue =
        xQueueCreate(client_queue_size, sizeof(struct ClientsMsg*));
    xPollingStartSemaphore = xSemaphoreCreateBinary();
    xSemaphore = xSemaphoreCreateBinary();
    xPollDataSemaphore = xSemaphoreCreateBinary();
    xTaskCreate(&polling_task, "polling_task", 4096, this,
                (configMAX_PRIORITIES - 2), NULL);
    xTaskCreate(&process_request_task, "process_request_task", 4096, this,
                (configMAX_PRIORITIES - 1), NULL);
    //xTaskCreate(&stream_cb_task, "stream_cb_task", 4000, this,
    //            (configMAX_PRIORITIES - 1), NULL);
    xTaskCreate(&stream_cb_immed_task, "stream_cb_immed_task", 4096, this,
                (configMAX_PRIORITIES - 1), NULL);
    return 0;
}

//入口
int WebSocketService::start(void) {
    ws_server_start();
    xTaskCreate(&create_ws_server_task, "create_ws_server_task", 4096, this,
                (configMAX_PRIORITIES - 2), NULL);
    xTaskCreate(&create_ws_session_task, "create_ws_session_task", 4096, this, (6),
                NULL);
    mStatus = WEB_SOCKET_STARTED;
    return 0;
}

int WebSocketService::postData(uint8_t* buf, int len) {
    int sendLen = 0;
    const static char* TAG = "postData";
    if (mStatus != WEB_SOCKET_STARTED) {
        return -1;
    }
    if ((poll_data_pos + len) > MSG_SIZE) {
        // send to polling_task
        ESP_LOGI(TAG, "postData dispatch msg start:%d E", poll_data_pos);
        xSemaphoreGiveFromISR(xPollDataSemaphore, &xHigherPriorityTaskWoken);
        poll_data_pos = 0;
        ESP_LOGI(TAG, "postData dispatch msg X");
    } else {
        memcpy(poll_data + poll_data_pos, buf, len);
        poll_data_pos += len;
        sendLen = len;
        // ESP_LOGI(TAG, "postData dispatch msg prepare:%d", poll_data_pos);
    }
    return sendLen;
}

int WebSocketService::postDataImmediately(uint8_t* buf, int len) {
    if (mStatus != WEB_SOCKET_STARTED || len <= 0
        || !WifiManager::getInstance()->getWifiConnected()) {
        printf("post fail: mStatus:%d len:%d wifi:%d\n", mStatus, len, WebService::getInstance()->getWifiStatus());
        return -1;
    }
    if (len > MSG_SIZE) {
        printf("data too long");
        len = MSG_SIZE;
    }
    if (gWebSocketConn) {
        ws_server_send_text_all_from_callback((char*)buf, len);  // pang
    }
    return len;
}
